package com.zhang.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.zhang.app.func.DimSinkFunction;
import com.zhang.app.func.TableProcessFunction;
import com.zhang.bean.TableProcess;
import com.zhang.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.table.runtime.typeutils.MapDataSerializer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @title: 动态分流
 * @author: zhang
 * @date: 2022/5/4 21:12
 * 数据流： web/app -> Nginx -> 业务服务器 -> Mysql(BinLog) -> Kafka(ODS) -> FlinkApp -> Phoenix(DIM)
 * 程序： MOck -> Nysql(Binlog) -> maxwell -> Kafka(ZK) -> DimApp -> Phoenix(HBase、HDFS、ZK)
 */
public class DimApp {
    public static void main(String[] args) throws Exception {
        //TODO 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        /*env.setStateBackend(new HashMapStateBackend());
        env.enableCheckpointing(5000L);
        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        env.getCheckpointConfig().setCheckpointStorage("hdfs:hadoop102:8020/");*/

        //TODO 读取kafka topic_db 创建业务数据流
        //TODO 过滤掉非JSON格式的数据，并将其写入侧输出流
        SingleOutputStreamOperator<JSONObject> kafkaDS = env
                .addSource(MyKafkaUtil.getKafkaConsumer("topic_db", "dim_app"))
                .process(new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(value);
                            out.collect(jsonObject);
                        } catch (Exception e) {
                            ctx.output(new OutputTag<String>("Dirty") {
                            }, value);
                        }
                    }
                });


        //TODO 使用FlinkCDC读取Mysql中的配置信息
        //TODO 将配置信息处理成广播流
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop103")
                .port(3306)
                .username("root")
                .password("000000")
                .databaseList("gmall_realtime_2022")
                .tableList("gmall_realtime_2022.table_process_dev")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<>("map-state", Types.STRING, Types.POJO(TableProcess.class));
        BroadcastStream<String> broadcastStream = env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MysqlSource")
                .broadcast(mapStateDescriptor);


        //TODO 连接主流和广播流
        //TODO 根据广播流数据处理主流数据
        SingleOutputStreamOperator<JSONObject> hbaseDS = kafkaDS
                .connect(broadcastStream)
                .process(new TableProcessFunction(mapStateDescriptor));

        //TODO 将数据写入到Phoenix中
        hbaseDS.print(">>>>>>>>>");
        hbaseDS.addSink(new DimSinkFunction());
        //TODO 启动任务
        env.execute();
    }
}
